package com.teemor.common.kafka.consumers;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.teemor.common.bean.model.EmailModel;
import com.teemor.common.bean.utils.IPInfoUtils;
import com.teemor.common.bean.utils.kafka.BaseKafkaProducer;
import com.teemor.common.bean.utils.redis.RedisUtils;
import com.teemor.common.kafka.fc.IBlogFc;
import com.teemor.common.kafka.fc.IEmailFc;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Description:
 * @author zhoulk
 * @date: 2020/3/2
 */
@Component
public class BaseConsumer {

    private static final Logger LOGGER = LoggerFactory.getLogger(BaseConsumer.class);

    @Autowired
    IEmailFc iEmailFc;
    @Autowired
    IBlogFc iBlogFc;
    @Autowired
    private RestTemplate restTemplate;

//    @KafkaListener(containerFactory = "simpleFactory", topics = "sms-topic")
    public void smsConsumer() {

    }

//    @KafkaListener(containerFactory = "simpleFactory", topics = "email-topic")
    public void emailConsumer(List<ConsumerRecord<String, EmailModel>> records, Acknowledgment ack) {
        for (ConsumerRecord<String, EmailModel> record : records) {
            ack.acknowledge();
            EmailModel emailModel = record.value();
        }

    }
    @KafkaListener(containerFactory = "simpleFactory", topics = "topic-iprecord")
    public void recordIpConsumer(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        for (ConsumerRecord<String, String> record : records) {
            Map<String, String> map = JSON.parseObject(record.value(),new TypeReference<Map<String, String>> (){});
            LOGGER.info("网站访问记录：" + record.toString());
            String ip = map.get("ip");
            if (!RedisUtils.exists(ip)) {
                if (RedisUtils.setStringNx(ip, "")) {
                    RedisUtils.setTime(ip, 60 * 60 * 6);
                    iBlogFc.addIprecord(map);
                }
            }
            ack.acknowledge();
        }

    }

    @KafkaListener(containerFactory = "simpleFactory", topics = "topic-footprint")
    public void footprintConsumer(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        for (ConsumerRecord<String, String> record : records) {
            LOGGER.info("文章访问记录：" + record.toString());
            Map<String, String> map = JSON.parseObject(record.value(), new TypeReference<Map<String, String>> (){});
            iBlogFc.addFootprint(map);
            ack.acknowledge();
        }
    }
    @KafkaListener(containerFactory = "elkFactory", topics = "topic-iprecord")
    public void elkRecordFootprintConsumer(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        for (ConsumerRecord<String, String> record : records) {
            LOGGER.info("文章访问记录：" + record.toString());
            Map<String, String> map = JSON.parseObject(record.value(), new TypeReference<Map<String, String>> (){});
            map.put("date", DateFormatUtils.format(new Date(Long.valueOf(map.get("date"))), "yyyy/MM/dd HH:mm:ss"));
            map.put("addr", IPInfoUtils.getIpaddr(restTemplate, map.get("ip")));
            BaseKafkaProducer.sendMessage(new ProducerRecord<>("topic-record_footprint_es", JSON.toJSONString(map)));
            ack.acknowledge();
        }
    }
}
